package com.Parser;

import com.Pojo.FileBeat;
import com.Pojo.NginxLog;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Locale;

/**自定义Filebeat-nginx解析器
 * nginx日志配置（非默认）：
 * log_format main
 *         '[$time_local] | $remote_addr | $remote_user | "$request" |'
 *         ' $status | $body_bytes_sent | "$http_referer" |'
 *         ' $connection | $upstream_addr |'
 *         ' "$http_user_agent" | "$http_x_forwarded_for" | "$uri" |'
 *         ' "$http_host" | $request_time | $upstream_response_time | $upstream_connect_time | $upstream_header_time ';
 *
 * filebeat.yml配置的input(fields是可选配项，日志解析类型优先级最高)：
 * filebeat.inputs:
 * - type: log
 *   paths:
 *     - /var/log/nginx/access.log
 *   fields:
 *     level: info
 *     review: 1
 *     agent_ip: 10.52.3.18
 *     type: |
 *
 */
public class FilebeatJSONKeyValueNginxDeserializationSchema implements KafkaDeserializationSchema<String> {

    private static final long serialVersionUID = 1509391548173891955L;

    private final boolean isJSON;

    public FilebeatJSONKeyValueNginxDeserializationSchema(boolean isJSON) {
        this.isJSON = isJSON;
    }

    @Override
    public boolean isEndOfStream(String jsonNodes) {
        return false;
    }

    @Override
    public String deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
        //Flink解析器固定返回String类型 提醒：接口是泛型，具体返回类型可自己定义
        String node = "" ;


        if (record.value() != null) {
            //使用fastjson解析日志处理，获取实际的Filebeat日志然后转为json字符串
            FileBeat fileBeat = JSON.parseObject(record.value(), FileBeat.class);
//            System.out.println(fileBeat.toString()); //检查数据
            Boolean isagentkey = false; //记录filebeat是否传agent_ip的fileds
            String agent_ip = ""; //默认设置agent_ip为null，若客户端有传agent_ip则插入数据库
            Boolean isJsonV2 = this.isJSON;
            if(fileBeat.getFields() != null){//先判断Fields是否存在

//                System.out.println("fields配置："+fileBeat.getFields()); //输出配置内容

                JSONObject result = JSONObject.parseObject(fileBeat.getFields()); //转为JSON类
                isagentkey = result.containsKey("agent_ip"); //判断是否存在agent_ip的key
                if (isagentkey){// 设置agent_ip
                    agent_ip = result.getString("agent_ip"); //获取filebeat所在服务器IP
                }else {
                    agent_ip = null;
                }
                if (result.containsKey("type")){// 若filebeat有配置type的数据解析类型-message，则以配置的解析方式为主
//                    System.out.println(result.getString("type").toLowerCase().equals("json") );
                    if(result.getString("type").toLowerCase().equals("json") ){
                        isJsonV2 = true;
                    }else {
                        isJsonV2 = false;
                    }
                }
            }else {
                agent_ip = null;
            }

//            System.out.println("agent_ip:"+agent_ip); //检查数据-作为日志输出列也行
            if(isJsonV2){
                //System.out.println(fileBeat.getMessage()); //检查数据
                NginxLog nginxLog = JSON.parseObject(fileBeat.getMessage(), NginxLog.class);
                nginxLog.setAgent_ip(agent_ip);
                //JSON字符串类型转为byte[]，写入node
                node = JSONObject.toJSONString(nginxLog);
            }else {
                String message = fileBeat.getMessage();
                //nginx日志切割，分隔符" | "
                NginxLog nginxLog2 = new NginxLog();
                //特殊字符|需要避免转义
                String [] message_1 = message.split(" \\| ");
                if (message_1.length >= 17){        // 修改长度限制为大于等于17.避免限死。只是不处理后面新增的日志列
                    //清除时间的括号，方便解析
                    String datestring = message_1[0].replace("[","").replace("]","");
                    //支持nginx-time_local、time_iso8601两种日期格式解析
                    try{
                        //T字符需要单引号括住
                        LocalDateTime dateTime = LocalDateTime.parse(datestring, DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss+08:00"));
                        nginxLog2.setTime_iso8601(dateTime);
                    } catch (Exception e) {
                        // 使用Locale.ENGLISH确保月份是以英文解析
                        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("dd/MMM/yyyy:HH:mm:ss Z", Locale.ENGLISH);
                        LocalDateTime dateTime = LocalDateTime.parse(datestring, formatter);
                        nginxLog2.setTime_iso8601(dateTime);
                    }
                    nginxLog2.setRemote_addr(message_1[1]);
                    nginxLog2.setRemote_user(message_1[2]);
                    nginxLog2.setRequest(message_1[3]);
                    nginxLog2.setStatus(message_1[4]);
                    nginxLog2.setBody_bytes_sent(message_1[5]);
                    nginxLog2.setHttp_referer(message_1[6]);
                    nginxLog2.setConnection(message_1[7]);
                    nginxLog2.setUpstream_addr(message_1[8]);
                    nginxLog2.setHttp_user_agent(message_1[9]);
                    nginxLog2.setHttp_x_forwarded_for(message_1[10]);
                    nginxLog2.setUri(message_1[11]);
                    nginxLog2.setHttp_host(message_1[12]);
                    nginxLog2.setRequest_time(message_1[13]);
                    nginxLog2.setUpstream_response_time(message_1[14]);
                    nginxLog2.setUpstream_connect_time(message_1[15]);
                    nginxLog2.setUpstream_header_time(message_1[16]);
                    nginxLog2.setAgent_ip(agent_ip);
                    node = JSONObject.toJSONString(nginxLog2);

                }else {
                    System.out.println(message_1.length);
                    for(String msg:message_1){
                        System.out.println(msg);
                    }
                    throw new Exception("截取长度不一致，解析nginx日志失败-分隔符-( | )");
                }
                //node = message; //调试-查看原数据
            }

        }

        return node;
    }


    @Override
    public TypeInformation<String> getProducedType() {
        return TypeExtractor.getForClass(String.class);
    }
}
